package com.dxf.bigdata.D04_spark

import org.apache.commons.logging.{Log, LogFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD方法主要有 转换 和 行动
 *   转换 如  Map , flatMap
 *   行动 collect
 *
 *
 *   RDD算子:   RDD的方法就是算子 --> 对数据的操作称为算子
 *
 *   RDD算子 => 转换算子 Map , 行动算子  collect
 *
 *   ===========================
 *
 *
 *   Map  => 处理数据,一个处理完了处理下一个 ==> 效率肯定底 (对比字节流) ==> 处理方式 缓存区
 *
 *
 *
 *  =====================================
 *
 *
 *  FlatMap 扁平映射
 *
 *
 */
object T12_FlatMap {
  private val log: Log = LogFactory.getLog(T08_RDD使用_从日志中读取uri.getClass)

  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)


    //TODO 创建RDD
    /*  val rdd: RDD[List[Int]] = sc.makeRDD(List(List(1, 2), List(3, 4)))
      val flatMap: RDD[Int] = rdd.flatMap(
        list => {
          list
        }
      )

      flatMap.foreach(println(_))*/


    val rdd3: RDD[String] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    /**
     * flatMap ()    传入一个可迭代的数据 ,返回一个可迭代的数据
     *
     *
     */
    val flatRdd: RDD[String] = rdd3.flatMap(str => {
      str.split(" ")
    })


    flatRdd.foreach(println(_))


    val rdd= sc.makeRDD(List(List(1, 2),3 , List(4, 5)))


    val rddResult: RDD[Int] = rdd.flatMap(list => {
      list match {
        case x: Int => List(x)
        case l: List[Int] => l
      }
    })

    rddResult.foreach(println(_))











    //TODO 关闭环境
    sc.stop()


  }
}
